#include <config.h>
#include <core.h>

volatile  msgd_uint_t msgd_current_msec;
block_queue_t         msgd_posted_events_queue;
extern msgd_int_t     msgd_accept_disabled;

void msgd_event_accept(msgd_event_t *ev)
{
  int               save_err;
  msgd_uint_t       available , socketlen;
  msgd_listening_t  *ls;
  msgd_connection_t *c;
  msgd_event_t      *rev , *wev;
  u_char            sa[MAX_SOCKET_LEN];
  msgd_int_t        s;
  pool_t            *pool;
 
  c  = ev->data;
  ls = c->listening; 

  available = 1;

  DEBUG("accept on %s:%s ",ls->addr_text,ls->port_text);

  do{
    socketlen = MAX_SOCKET_LEN;

    s = accept(c->fd,(struct sockaddr *)sa,&socketlen);

    if(s == -1){
       save_err = errno;
       if(save_err == EAGAIN){
          ERROR("accept() cant ready (EAGAIN)");
          continue;
       }
      
       if(save_err == ECONNABORTED){
          ERROR("accept() failed ECONNABORTED");
          continue;
       }
 
       DEBUG("accept failed available:%d",available);
       return;
    }

    msgd_accept_disabled  = global.connections_n / 8 
                               - global.free_connections_n;

    c = msgd_get_connection(s);
    if(c == NULL){
      if(close(s) == -1){
         ERROR("close socket %d failed",s);
      } 
      return ;
    }

    pool = create_pool(CONN_POLL_SIZE);
    if(pool == NULL){
      msgd_close_accepted_connection(c);
      ERROR("mem not enough");
      return;
    }

    rev = c->read;
    wev = c->write;

    memset(c,0,sizeof(msgd_connection_t));
   
    c->read = rev;
    c->write  = wev; 
    c->fd   = s;
    c->listening = ls;
    c->pool = pool;
    
    c->sockaddr = pool_palloc(pool,socketlen);
    if(c->sockaddr == NULL){
      msgd_close_accepted_connection(c);
      ERROR("mem not enough");
      return;
    }
    memcpy(c->sockaddr,sa,socketlen);
    c->socklen = socketlen;

    DEBUG("accept on %s:%s fd:%d ",ls->addr_text,ls->port_text,s);

    ls->handler(c);

  }while(available);
}



void msgd_empty_write_handler(msgd_event_t *wev)
{
   DEBUG("empty write handler %p",wev);
   return ;
}

msgd_int_t msgd_handle_read_event(msgd_event_t *rev)
{
    if(!rev->active){
       if(msgd_add_event(rev,MSGD_READ_EVENT,MSGD_CLEAR_EVENT) == MSGD_ERROR)
       {
           return MSGD_ERROR;
       }
    }

    return MSGD_OK;
}


void msgd_handle_posted_event(msgd_event_t **queue)
{
    msgd_event_t *ev;

    for(;;){

        ev  = (msgd_event_t *)*queue;
     
        if(ev == NULL)
          return;

        DEBUG("posted event:%p queue:%p",ev,queue);
      
        msgd_delete_posted_event(ev);
 
        ev->handler(ev); 
    }
}


void msgd_time_update(void)
{
  time_t       sec;
  msgd_uint_t  msec;
  struct timeval tv;
  
  gettimeofday(&tv,NULL); 

  sec = tv.tv_sec;
  msec = tv.tv_usec / 1000;

  msgd_current_msec = sec * 1000 + msec;
}
